package org.bing.zion.filter.serializer;

import java.io.IOException;
import java.nio.ByteBuffer;

import org.bing.engine.common.logging.Log;
import org.bing.engine.common.logging.LogFactory;
import org.bing.zion.core.ProtocolFilterAdapter;
import org.bing.zion.core.ProtocolFilterChain;
import org.bing.zion.core.Session;


public class SimpleSerializerFilter extends ProtocolFilterAdapter {
	private static final String SESSION_KEY_HIS_BYTES = "KEY.SIMPLESERIALIZERFILTER.HIS.BYTES";
	private static final Log logger = LogFactory.getLog(SimpleSerializerFilter.class);
	private static final SimpleReflectDecoder decoder = new SimpleReflectDecoder();
	private static final int MAX_OBJECT_SIZE = 1024 * 256;

	private byte[] mergeBytes(Session session, ByteBuffer buf) {
		byte[] total = (byte[]) session.getAttribute(SESSION_KEY_HIS_BYTES);
		if (buf == null) {
			if (total == null) {
				return new byte[0];
			} else {
				return total;
			}
		} else {
			byte[] cbs = getBytes((ByteBuffer) buf.flip());
			// if (total != null)
			// logger.info("HIS " + new String(total));
			// if (cbs != null)
			// logger.info("CBS " + new String(cbs));
			if (total == null) {
				return total = cbs;
			} else {
				byte[] tmp = new byte[total.length + cbs.length];
				if (total.length > 0) {
					System.arraycopy(total, 0, tmp, 0, total.length);
				}
				if (cbs.length > 0) {
					System.arraycopy(cbs, 0, tmp, total.length, cbs.length);
				}
				total = tmp;
			}
		}
		if (total != null && total.length > MAX_OBJECT_SIZE) {
			logger.error("Object's size may over 256K! ");
			logger.error(new String(total));
			throw new RuntimeException("Object's size may over 256K!");
		}
		return total;
	}

	private String readObjectString(Session session, ByteBuffer buf) {
		byte[] total = mergeBytes(session, buf);
		// logger.info("TOT " + new String(total));
		int bInx = -1;
		int eInx = -1;
		int num = 0;
		boolean f = false;
		// check history
		int limit = total.length;
		for (int i = 0; i < limit; i++) {
			byte c = total[i];
			if (c == '<') { // begin '<...', or end </...>
				if ((i + 1) < limit && '/' == total[i + 1]) {
					if ((i + 2) < limit && num == 0) {
						int pos = index(total, (byte) '>', i + 2);
						if (pos > 0) {
							eInx = pos + 1;
						}
						break;
					} else {
						num--;
					}
				} else if (!f) {
					bInx = i;
					f = true;
				} else {
					num++;
				}
			} else if (c == '/') { // end '/>'
				if ((i + 1) < limit && '>' == total[i + 1]) {
					if ((i + 2) < limit && num == 0) {
						eInx = i + 2;
						break;
					} else {
						num--;
					}
				}
			}
		}
		if (bInx >= 0 && eInx > 0) {
			String str = new String(total, bInx, eInx - bInx);
			byte[] tmp = new byte[total.length - eInx];
			System.arraycopy(total, eInx, tmp, 0, tmp.length);
			session.setAttribute(SESSION_KEY_HIS_BYTES, tmp);
			return str;
		} else {
			session.setAttribute(SESSION_KEY_HIS_BYTES, total);
			return null;
		}
	}

//	private static volatile Long read = 0L;
//	private static volatile Long write = 0L;

	@Override
	public void messageReceived(ProtocolFilterChain chain, Session session, Object msg) throws IOException {
		if (msg instanceof ByteBuffer) {
			// logger.info("Do simple deserialize ... ");
			String str = readObjectString(session, (ByteBuffer) msg);
			while (str != null) {
				// logger.info("OKK " + str);
				Object obj = decoder.read(str);
				if (obj != null) {
					chain.next().messageReceived(chain, session, obj);
//					synchronized (read) {
//						read++;
//						if (read % 1 == 0) {
//							System.out.println("READ to " + read);
//						}
//					}
				}
				str = readObjectString(session, null);
			}
		} else {
			chain.next().messageReceived(chain, session, msg);
		}
	}

	@Override
	public void messageSent(final ProtocolFilterChain chain, final Session session, final Object msg)
			throws IOException {
		ByteBufferWriter writer = new ByteBufferWriter() {
			@Override
			public void flush() throws IOException {
				buffer.flip();
				parentMessageSent(chain, session, buffer);
				buffer = bufferPool.obtain();
			}
		};
		writer.write(msg);
		writer.close();

//		synchronized (write) {
//			write++;
//			if (write % 1 == 0) {
//				System.out.println("WRITE to " + write);
//			}
//		}
	}

	private void parentMessageSent(ProtocolFilterChain chain, Session session, Object msg) throws IOException {
		super.messageSent(chain, session, msg);
	}

}
